package MapReduce.Demo12_ProductAndOrder.mapJoin;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.yarn.webapp.hamlet2.Hamlet;
import utils.JobSubmit;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

/**
 * @Author lixinlei
 * @Date 2023/3/23 9:15
 */
public class ProductAndOrderMapJoinApp {

    public static class PAOMJMapper extends Mapper<LongWritable, Text,Text, NullWritable>{

        Text outKey = new Text();

        //初始化商品信息的map，key是商品id value是商品名
        Map<String,String> productMap = new HashMap<>();

        @Override
        protected void setup(Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
            //获取driver中设置的cacheFile 的第一个文件 -- 商品文件
            URI productFile = context.getCacheFiles()[0];
            //获取商品文件的路径
            String productFilepath = productFile.getPath();

            //使用商品文件创建出一个文件字符输入流
            FileReader fr = new FileReader(productFilepath);
            //把文件字符输入流封装成缓冲流，目的方便后面读取每行的数据
            BufferedReader br = new BufferedReader(fr);

            String line = "";
            //读取每行数据，把商品信息封装成Map中的键值对
            while((line = br.readLine())!=null){
                String[] fields = line.split("\t");
                productMap.put(fields[0],fields[1]);
            }

            //关闭流
            if(br!=null){
                br.close();
            }

            if(fr!=null){
                fr.close();
            }
        }

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] fields = line.split("\t");

            outKey.set(fields[0]+"\t"+fields[1]+"\t"+productMap.get(fields[1])+"\t"+fields[2]);

            context.write(outKey,NullWritable.get());
        }
    }

    public static void main(String[] args) throws Exception {

//        Configuration conf = new Configuration();
//        Job job = Job.getInstance(conf);
//
//        job.setMapOutputKeyClass(Text.class);
//        job.setMapOutputValueClass(NullWritable.class);
//
//        job.setJarByClass(ProductAndOrderMapJoinApp.class);
//        job.setMapperClass(PAOMJMapper.class);
//
//
//        Path outputPath = new Path(args[1]);
//        FileSystem fs = FileSystem.get(conf);
//
//        if (fs.exists(outputPath)) {
//            fs.delete(outputPath,true);
//        }
//
//
//        FileInputFormat.setInputPaths(job,new Path(args[0]));
//        FileOutputFormat.setOutputPath(job,outputPath);
//
//
//        job.setNumReduceTasks(0);
//
//        URI[] uris = {new URI(args[2])};
//        job.setCacheFiles(uris);
//
//        job.waitForCompletion(true);


        URI[] uris = {new URI(args[2])};
        JobSubmit.submitBaseJob(
                ProductAndOrderMapJoinApp.class,
                args,
                0,
                uris
        );

    }

}
